package KafkaApps;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.header.internals.RecordHeaders;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.apache.kafka.common.serialization.LongSerializer;


import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.ExecutionException;


public class ProducerPiece {
    final static private String TOPIC = "testTrans";
    private final static String FILEPATH = "trans_file/producer_file/ten-million.txt";

    private static Producer<Long, byte[]> createProducer(){
        Properties set = new Properties();
        set.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "master:9092");
        set.put(ProducerConfig.CLIENT_ID_CONFIG, "ProducerFiles");
        set.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, LongSerializer.class.getName());
        set.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,ByteArraySerializer.class.getName());
        return new KafkaProducer<>(set);
    }

    private static List<byte[]>  readFileCut(File file) throws IOException {
        FileInputStream fileInputStream = null;
        List<byte[]> imgList = new ArrayList<>();
        try {
            fileInputStream  = new FileInputStream(file);
            while(true) {
                byte[] fileBytes = new byte[1024 * 10 ];
                if (fileInputStream.read(fileBytes) == -1) break;
                imgList.add(fileBytes);
            }

            return imgList;
        } catch (FileNotFoundException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            fileInputStream.close();
        }
        return  null;
    }

    private static boolean sendMsg(List<byte[]> fileList, Producer<Long, byte[]> producer){
        long time = System.currentTimeMillis();

        int i = 1;
        for(byte[] bytes:fileList){
            RecordHeaders headers = new RecordHeaders();
            headers.add("number", Integer.toString(i).getBytes());
            final ProducerRecord<Long, byte[]> producerRecord =
                    new ProducerRecord<Long, byte[]>(TOPIC, i%12, time, bytes, headers);

            try {
                RecordMetadata metadata = producer.send(producerRecord).get();
                System.out.printf("发送文件%d成功！meta(partition=%d, offset=%d) \n", i++
                        ,metadata.partition(),metadata.offset());
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
        }
         {
            producer.flush();
            producer.close();
        }
        return false;
    }



    static void runProducer(){
        File file = new File(FILEPATH);
        final Producer<Long, byte[]> producer = createProducer();
        try {
            sendMsg(readFileCut(file), producer);
        } catch (IOException e) {
            e.printStackTrace();
        }

    }

    public static void main(String... args) throws Exception {
        runProducer();
    }
}
